/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2013_06_06			*/
/*	File:		MrUtl.cpp			*/
/********************************************************/

#include "MrTask.h"

void MrTkQuit(int iSignal)
{
	MR_ERROR("Stopped by signal[%d]\n", iSignal);

	exit(MR_FAILURE);
}

void MrTkHelp(char* pszProc, char* pszVersion)
{
	MR_PRINT("Usage: %s [options]\n", pszProc);
	MR_PRINT("Valid options are:\n");
	MR_PRINT("[*] -c or --conf:       set configure.\n");
	MR_PRINT("[0] -p or --pool:       set pool total.\n");
	MR_PRINT("[0] -d or --debug:      set debug mode.\n");
	MR_PRINT("[0] -h or --help:       show help.\n");
	MR_PRINT("[0] -v or --version:    show version.\n");
	MR_PRINT("Copyright (C) 2012 Gong Li Bin	\n");
	MR_PRINT("%s Build on %s %s.\n", pszVersion, __DATE__, __TIME__);

	return;
}

int MrTkConfig(char* pszConf, char* pszSql, char* pszPort)
{
	CMrDb CDb;
	MRCONFIG_S stConf;
	MRPROKV_S* pstKv = NULL;
	int iReturn = MR_SUCCESS;
	char szPacket[MR_PACKET] = { 0 };

	memset(&stConf, '\0', sizeof(MRCONFIG_S));
	if (MR_SUCCESS == MrConfig(pszConf, &stConf)) {
		if (NULL != (pstKv = MrProKvMalloc(MR_BUFFER, MR_PACKET))) {
			if (MR_SUCCESS == CDb.MrDbConnect(stConf.m_szHost, stConf.m_szPort)) {
				sprintf(szPacket, pszSql, stConf.m_szDbase, stConf.m_szTable, stConf.m_szID);
				if (CDb.MrDbSend(MR_DEF_SELT, szPacket) > 0) {
					if (MR_SUCCESS == CDb.MrDbRecv(pstKv)) {
						memcpy(pszPort, (pstKv + 0)->m_puszValue, (pstKv + 0)->m_iLength);
					}
					else {
						MR_ERROR("Failed to MrDbRecv\n");
						iReturn = MR_FAILURE;
					}
				}
				else {
					MR_ERROR("Failed to MrDbSend\n");
					iReturn = MR_FAILURE;
				}
				CDb.MrDbClose();
			}
			else {
				MR_ERROR("Failed to MrDbConnect\n");
				iReturn = MR_FAILURE;
			}
			MrProKvFree(&pstKv, MR_BUFFER);
		}
		else {
			MR_ERROR("Failed to MrProKvMalloc\n");
			iReturn = MR_FAILURE;
		}
	}
	else {
		MR_ERROR("Failed to MrConfig\n");
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

void MrTkResponse(CMrUdp& rCUdp)
{
	CMrPro CPro;
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };

	CPro.MrProResponse(uszHead);
	rCUdp.MrNetSend(uszHead, MR_PRO_HEAD);

	return;
}

UCHAR MrTkOutput(CMrSet& rCHub, MRPARA_S& rstOut, CMrPro& rCPro, UCHAR* puszPacket)
{
	CMrMd5 CMd5;
	CMrTcp CTcp;
	UINT uiHub = 0;
	CMrPro CProSend;
	CMrPro CProRecv;
	UINT uiCount = 0;
	USHORT usTotal = 0;
	char* pszUrl = NULL;
	struct timeval stTime;
	char szUrl[MR_URL] = { 0 };
	char szKey[MR_VALUE] = { 0 };
	UCHAR uszMd5[MR_MD5] = { 0 };
	UCHAR ucReturn = MR_DEF_SUCC;
	char szSample[MR_BUFFER] = { 0 };
	UCHAR uszPacket[MR_PACKET] = { 0 };
	MRPROKV_S stKv[MR_MR_VAL] = {{MR_MR_KEY, 0, (UCHAR*)szKey}, {MR_MR_VAL, 0, NULL}};
	MRPROHEAD_S stHead = {
		MR_PRO_FLAG,
		MR_PRO_1,
		MR_PRO_DATA,
		MR_PRO_0,
		MR_PRO_Y,
		MR_PRO_0,
		MR_DEF_EHUB,
		MR_DEF_INST,
		MR_PRO_0,
		MR_DEF_SUCC,
		MR_MR_VAL,
		MR_PRO_0,
	};

	while (uiCount < rstOut.m_uiSize && ((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_iLength > 0) {
		uiHub = uiCount;

		/* sample */
		srand((UINT)time(NULL));
		gettimeofday(&stTime, NULL);
		sprintf(szSample, "%d%ld%ld%d", getpid(), stTime.tv_sec, stTime.tv_usec, rand());
		memcpy(szSample + strlen(szSample), ((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_puszValue, MR_BUFFER - strlen(szSample));

		/* MD5key */
		CMd5.MrMd5Run((UCHAR*)szSample, MR_BUFFER, uszMd5);
		sprintf(szKey, "%llu", *(ULONGLONG*)uszMd5 ^ *(ULONGLONG*)(uszMd5 + sizeof(ULONGLONG)));

		/* packet */
		stKv[MR_PRO_1].m_puszValue = uszMd5;
		stKv[MR_PRO_0].m_iLength = strlen(szKey);
		stKv[MR_PRO_1].m_iLength = ((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_iLength;

		CProSend.MrProPutBody(NULL);
		CProSend.MrProInduce(&stHead);
		CProSend.MrProPutLength(MrProKvOutput(stKv, MR_MR_VAL, &uszPacket[MR_PRO_HEAD], MR_PACKET - MR_PRO_HEAD));
		CProSend.MrProCreate(uszPacket, MR_PACKET);

		/* output */
		do {
			pszUrl = (char*)rCHub.MrSetIndexGet(uiHub);
			memcpy(szUrl, rCHub.MrSetIndexGet(uiHub), MR_URL);

			if (MR_SUCCESS == CTcp.MrTcpConnect(szUrl)) {
				CTcp.MrTcpSetLinger(0);
				CTcp.MrTcpSetNoDelay();
				if (CTcp.MrNetSend(uszPacket, MR_PRO_HEAD + CProSend.MrProGetLength() - stKv[MR_PRO_1].m_iLength) > 0) {
					stKv[MR_PRO_1].m_puszValue = ((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_puszValue;
					if (stKv[MR_PRO_1].m_iLength == CTcp.MrNetSendAll(stKv[MR_PRO_1].m_puszValue, stKv[MR_PRO_1].m_iLength)) {
						if (MR_PRO_HEAD == CTcp.MrNetRecv(uszPacket, MR_PRO_HEAD)) {
							CProRecv.MrProParse(uszPacket, MR_PRO_HEAD);
							ucReturn = CProRecv.MrProGetResult();
						}
						else {
							MR_ERROR("Failed to MrNetRecv\n");
							ucReturn = MR_DEF_RECV;
						}
					}
					else {
						MR_ERROR("Failed to MrNetSend\n");
						ucReturn = MR_DEF_SEND;
					}
				}
				else {
					MR_ERROR("Failed to MrNetSend\n");
					ucReturn = MR_DEF_SEND;
				}
				CTcp.MrNetClose();
			}
			else {
				MR_ERROR("Failed to MrTcpConnect\n");
				ucReturn = MR_DEF_CONN;
			}

			if (++ uiHub == rCHub.MrSetGetCount()) {
				uiHub = 0;
			}

			if (uiHub == uiCount) {
				break;
			}
		} while (MR_DEF_SUCC != ucReturn);

		/* reset */
		if (MR_DEF_SUCC == ucReturn) {
			usTotal += 1;
			((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_ucKey = MR_MR_LCL;
			memcpy(((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_puszValue, pszUrl, MR_URL);
			((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_iLength = MR_URL + stKv[MR_PRO_0].m_iLength;
			memcpy(((MRPROKV_S*)rstOut.m_pPara + uiCount)->m_puszValue + MR_URL, szKey, stKv[MR_PRO_0].m_iLength);
		}

		uiCount += 1;
	}

	if (usTotal > 0) {
		rCPro.MrProPutTotal(usTotal);
		rCPro.MrProPutLength(MrProKvOutput((MRPROKV_S*)rstOut.m_pPara, usTotal, puszPacket, MR_PACKET - MR_PRO_HEAD));
	}

	return ucReturn;
}
